package util;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.hazelcast.cache.impl.nearcache.NearCache;
import com.hazelcast.cache.impl.nearcache.NearCacheManager;
import com.hazelcast.cache.impl.nearcache.impl.DefaultNearCacheManager;
import com.hazelcast.client.HazelcastClient;
import com.hazelcast.client.config.ClientConfig;
import com.hazelcast.client.impl.HazelcastClientProxy;
import com.hazelcast.config.EvictionConfig;
import com.hazelcast.config.EvictionConfig.MaxSizePolicy;
import com.hazelcast.config.EvictionPolicy;
import com.hazelcast.config.NearCacheConfig;
import com.hazelcast.core.EntryEvent;
import com.hazelcast.core.IMap;
import com.hazelcast.map.listener.EntryAddedListener;
import com.hazelcast.map.listener.EntryRemovedListener;
import com.hazelcast.map.listener.EntryUpdatedListener;

/**
 * 在cluster数据更新后，过一个延迟时间后，near cache才能更新数据
 * 
 * @author wxu
 *
 */
public class NearCacheTest implements Runnable {

	public static final String MAP = "map";
	private static final Logger logger = LoggerFactory.getLogger(NearCacheTest.class);
	private IMap<String, String> testmap;
	public static final String SEMAPHORE_EVICT = "evict";
	public static final String SEMAPHORE_ACK = "ack";

	public void run() {
		ClientConfig config = new ClientConfig();
		config.setProperty("hazelcast.map.invalidation.batch.enabled", "true");
		config.setProperty("hazelcast.map.invalidation.batch.size", "100");
		config.setProperty("hazelcast.map.invalidation.batchfrequency.seconds", "1000");

		EvictionConfig evictionConfig = new EvictionConfig();
		evictionConfig.setMaximumSizePolicy(MaxSizePolicy.ENTRY_COUNT);
		evictionConfig.setEvictionPolicy(EvictionPolicy.LFU);
		evictionConfig.setSize(10000);
		NearCacheConfig nearCacheConfig = new NearCacheConfig()
				// 设置需要使用near Cache的map
				.setName(MAP)
				// 1 hour TTL
				.setTimeToLiveSeconds(60 * 60 * 1000)
				// 10 minutes max idle seconds
				.setMaxIdleSeconds(10 * 60 * 1000)
				// Near Cache Invalidation
				.setInvalidateOnChange(false)
				// eviction
				.setEvictionConfig(evictionConfig);
		config.addNearCacheConfig(nearCacheConfig);

		HazelcastClientProxy hzClient = (HazelcastClientProxy) HazelcastClient.newHazelcastClient(config);

		NearCacheManager nearCacheManager = hzClient.client.getNearCacheManager();

		testmap = hzClient.getMap(MAP);
		testmap.addEntryListener(new MyEntryListener(), true);
		// check map size
		while (true) {
			logger.info("testmap(1):{}, testmap(2):{}", testmap.get("1"), testmap.get("2"));
//			System.out.println(nearCacheManager.listAllNearCaches().size());
			NearCache<Object, Object> nearCache = nearCacheManager.getNearCache(MAP);
			if (nearCache != null) {
				System.out.println(nearCache.getNearCacheStats());
			}
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

	}

	private class MyEntryListener implements EntryAddedListener<String, String>, EntryRemovedListener<String, String>,
			EntryUpdatedListener<String, String> {
		@Override
		public void entryAdded(EntryEvent<String, String> event) {
			logger.info("testmap size:{} ,entryAdded: {}", testmap.size(), event);
			// testmap.evict(event.getKey());
//			 testmap.evictAll();
			// testmap.set(event.getKey(), event.getValue());

		}

		@Override
		public void entryRemoved(EntryEvent<String, String> event) {
			logger.info("entryRemoved: " + event);
		}

		@Override
		public void entryUpdated(EntryEvent<String, String> event) {
			logger.info("entryUpdated: " + event);
			// testmap.evict(event.getKey());
			// testmap.evictAll();

		}
	}

	public NearCacheTest() {
		new Thread(this, "reader1").start();
	}
}
